package com.study.flink.java.day04_async;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.http.NameValuePair;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;


import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.*;

public class BalanceToActivityBeanFunction extends RichMapFunction<String, ActivityBean> {

    // 外部接口地址
    private static final String requestUrl = "http://yan.taobiso.com/Handler/Api.ashx?type=GetBalance";
    private static final String UID = "225881";

    private Connection conn = null; //jdbc
    private CloseableHttpClient httpClient = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 创建MySQL连接
        conn = DriverManager.getConnection("jdbc:mysql://node02:3306/bigdata?characterEncoding=UTF-8", "linys", "123456");
        httpClient = HttpClients.createDefault();
    }

    // line example: u001,A1,2020-02-26 10:11:12,1,北京市
    // line example: u002,A2,2020-02-25 10:11:12,1,厦门市
    // line example: u003,A3,2020-02-25 20:44:55,1,泉州市
    // line example: u003,A4,2020-02-23 08:28:45,2,莆田市
    @Override
    public ActivityBean map(String line) throws Exception {
        System.out.println(line);
        String[] fields = line.split(",");
        String uid = fields[0];
        String aid = fields[1];

        // 转换为活动文字信息，查MySQL
        PreparedStatement preparedStatement = conn.prepareStatement("SELECT name FROM t_activities WHERE id = ?");
        preparedStatement.setString(1, aid);
        ResultSet resultSet = preparedStatement.executeQuery();
        String name = null;
        while (resultSet.next()) {
            name = resultSet.getString(1);
        }
        preparedStatement.close();

        //外部接口查询实时余额
        String ticks = DateUtil.format(new Date(System.currentTimeMillis()), DatePattern.NORM_DATETIME_PATTERN);
        String token = getToken(ticks);
        String balance = "";
        CloseableHttpResponse response = null;
        Map<String, String> paramMap = new HashMap<>();
        paramMap.put("uid", UID);
        paramMap.put("ticks", ticks);
        paramMap.put("nick", "");
        paramMap.put("token", token);
        try {
            HttpPost httpPost = new HttpPost(requestUrl);
            List<NameValuePair> content = new ArrayList<>();
            Iterator iterator = paramMap.entrySet().iterator();           //将content生成entity
            while(iterator.hasNext()){
                Map.Entry<String,String> elem = (Map.Entry<String, String>) iterator.next();
                content.add(new BasicNameValuePair(elem.getKey(), elem.getValue()));
            }
            if(content.size() > 0) {
                UrlEncodedFormEntity entity = new UrlEncodedFormEntity(content, "UTF-8");
                httpPost.setEntity(entity);
            }
            response = httpClient.execute(httpPost);
            int status = response.getStatusLine().getStatusCode();
            if(status == 200) {
                /**
                 * {
                 *   "Balance": "273.70",
                 *   "success": true,
                 *   "info": ""
                 * }
                 */
                String result = EntityUtils.toString(response.getEntity());
                System.out.println(result);
                JSONObject jsonObject = JSONObject.parseObject(result);
                balance = jsonObject.getString("Balance");
            }
        } finally {
            if(response != null)
                response.close();
        }

        String time = fields[2];
        int eventType = Integer.parseInt(fields[3]);
        String province = fields[4];
        return ActivityBean.of(uid, aid, name, time, eventType, province, balance);
    }

    @Override
    public void close() throws Exception {
        super.close();
        conn.close();
        httpClient.close();
    }

    private String getToken(String ticks) {
        StringBuffer query = new StringBuffer();
        query.append(UID);
        query.append(ticks);
        query.append("13C0D11FDC5A4EBBA3F5");
        String param = new String(query);
        byte[] data2 = null;
        try {
            data2 = param.getBytes("UTF-8");
        } catch (Exception e) {
        }
        String sign = DigestUtils.md5Hex(data2);
        return sign;
    }

}
